package cn.me.alphamq.broker.core.consumequeue;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.core.commitlog.CommitLogMMapFileModel;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueConsumeReq;
import cn.me.alphamq.common.model.broker.ConsumeQueueFileModel;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueIndexModel;
import cn.me.alphamq.common.model.broker.ConsumeQueueOffsetModel;
import cn.me.alphamq.common.util.FileNameUtil;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.util.AssertUtil;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 消费队列消费处理器
 *
 * @author f
 */
public class ConsumeQueueConsumeHandler {

    /**
     * 读取当前最新多条 ConsumeQueue 索引指向的数据内容
     *
     * @param consumeQueueConsumeReq 消费的请求
     * @return 如果 队列不存在 或 没有最新的 ConsumeQueue 索引单元数据，则返回 null
     */
    public List<Message> consume(ConsumeQueueConsumeReq consumeQueueConsumeReq) {
        String topicName = consumeQueueConsumeReq.getTopic();

        CommitLogTopicModel commitLogTopicModel = Container.fetchCommitLogTopicModelByTopic(topicName);
        AssertUtil.isNotNull(commitLogTopicModel, "topic [" + topicName + "] does not exist");

        String consumerGroup = consumeQueueConsumeReq.getConsumerGroup();
        Integer queueId = consumeQueueConsumeReq.getQueueId();
        Integer batchMsgNum = consumeQueueConsumeReq.getBatchMsgNum();

        // 获取 offset 表
        ConsumeQueueOffsetModel offsetMap = Container.getConsumeQueueOffsetModel();

        // 获取 消费者组表
        Map<String, Map<String, String>> consumerGroupMap = offsetMap.fetchConsumerGroupMapForcibly(topicName);

        // 获取 消费队列表
        List<ConsumeQueueFileModel> consumeQueueFileModelList = commitLogTopicModel.getQueueList();
        Map<String, String> consumeQueueMap = consumerGroupMap.computeIfAbsent(consumerGroup, item -> {
            // 如果是首次消费，就创建一个新的 消费队列表
            Map<String, String> newConsumeQueueMap = new HashMap<>();
            for (ConsumeQueueFileModel consumeQueueFileModel : consumeQueueFileModelList) {
                // 映射 00000000 这个新文件，起始的偏移量为 0
                newConsumeQueueMap.put(String.valueOf(consumeQueueFileModel.getId()),
                        FileNameUtil.generateConsumerOffset());
            }
            return newConsumeQueueMap;
        });

        // 获取 offset
        String fileNameAndOffsetInfo = consumeQueueMap.computeIfAbsent(String.valueOf(queueId),
                item -> FileNameUtil.generateConsumerOffset());
        int consumeQueueOffset = Integer.parseInt(fileNameAndOffsetInfo.split("#")[1]);

        // 如果 索引单元的 offset 大于或等于 本队列目前已经写入的最大 offset，则返回 null
        // TODO 考虑一下 consumeQueueFileModel 是否可能为 null
        ConsumeQueueFileModel consumeQueueFileModel = consumeQueueFileModelList
                .stream().filter(item -> item.getId().equals(queueId)).findFirst().orElse(null);
        AssertUtil.isNotNull(consumeQueueFileModel,
                "topic [" + topicName + "]'s queue [" + queueId + "] does not exist");
        if (consumeQueueOffset >= consumeQueueFileModel.getOffset().get()) {
            return new ArrayList<>();
        }

        // 获取对应的 MMap 文件模型
        ConsumeQueueMMapFileModel consumeQueueMMapFileModel =
                Container.getConsumeQueueMMapFileModelManager().get(topicName).get(queueId);
        // TODO 这个方法只能获取当前正在写入的 CommitLog 文件，如果是索引数据指向之前的 CommitLog 文件，则无法获取
        CommitLogMMapFileModel commitLogMMapFileModel = Container.getCommitLogMMapFileModelManager().get(topicName);
        List<byte[]> consumeQueueByteList = consumeQueueMMapFileModel.readBatchContent(consumeQueueOffset, batchMsgNum);

        List<Message> messageList = new ArrayList<>();
        for (byte[] content : consumeQueueByteList) {
            // 将 索引单元的数据 从 ConsumeQueue 文件中读出来
            ConsumeQueueIndexModel consumeQueueIndexModel = new ConsumeQueueIndexModel(content);

            // 如果没有数据，就不要再读取了
            int msgLength = consumeQueueIndexModel.getMsgLength();
            if (msgLength == 0) {
                break;
            }

            // 使用 索引单元的数据 从 CommitLog 文件中读取数据
            int msgOffset = consumeQueueIndexModel.getMsgOffset();
            byte[] body = commitLogMMapFileModel.readContent(msgOffset, msgLength);

            // 将 字节数组形式的 消息主体(byte[]) 封装成 消费者可以消费的 消息主体(Message)
            int retryTimes = consumeQueueIndexModel.getRetryTimes();
            messageList.add(new Message(body, msgOffset, retryTimes));
        }

        return messageList;
    }

    /**
     * 消费者消费完数据后，更新 ConsumeQueue 文件的 offset
     * <p>
     * TODO 是否需要提供一个批量 ACK 的接口
     *
     * @param topicName     topic 名称
     * @param consumerGroup 消费者组
     * @param queueId       队列 id
     * @return
     */
    public boolean ack(String topicName, String consumerGroup, Integer queueId) {
        // 获取 消费队列表
        Map<String, String> consumeQueueMap = Container.getConsumeQueueOffsetModel()
                .fetchConsumeQueueMap(topicName, consumerGroup);

        // 获取 ConsumeQueue 文件名称 和 消费 offset
        String fileNameAndOffsetInfo = consumeQueueMap.get(String.valueOf(queueId));
        String[] split = fileNameAndOffsetInfo.split("#");
        String consumeQueueFileName = split[0];
        int consumeQueueOffset = Integer.parseInt(split[1]);

        // 修改 消费详情
        consumeQueueMap.put(String.valueOf(queueId),
                consumeQueueFileName + "#" + (consumeQueueOffset + ConsumeQueueMMapFileModel.INDEX_UNIT_SIZE));
        return true;
    }

}
